use std::collections::HashMap;
use std::io::{self, Write};
use std::net::{TcpListener, TcpStream};
use std::os::unix::io::AsRawFd;
use std::sync::mpsc::sync_channel;
use std::sync::{Arc, Mutex};

pub mod ffi;

fn main() {
    let epfd = unsafe { ffi::epoll_create(1) };
    panic_if_error(epfd);

    // 监听本地的 3000 端口
    let listener = TcpListener::bind("127.0.0.1:3000").unwrap();
    listener.set_nonblocking(true).unwrap(); // 设置非阻塞
    let listener_fd = listener.as_raw_fd();
    let shared_listener = Arc::new(Mutex::new(listener));
    // TcpStream 连接容器
    let map: HashMap<i32, TcpStream> = HashMap::new();
    let shared_map = Arc::new(Mutex::new(map));

    // 监听事件
    let mut event = ffi::EpollEvent {
        events: (ffi::EPOLLIN | ffi::EPOLLRDHUP) as u32, // 监听 listener 上的读、对端关闭事件
        data: ffi::EpollData {
            fd: listener_fd, // 传入监听服务器的 FD
        },
    };

    let ret = unsafe { ffi::epoll_ctl(epfd, ffi::EPOLL_CTL_ADD, listener_fd, &mut event) };
    panic_if_error(ret);

    let n_workers = 4;

    let (sender, receiver) = sync_channel::<Event>(100);
    let shared_rx = Arc::new(Mutex::new(receiver));

    for i in 0..n_workers {
        let shared_rx = shared_rx.clone();
        let shared_listener = shared_listener.clone();
        let shared_map = shared_map.clone();
        let epfd = epfd.clone();
        std::thread::spawn(move || loop {
            let rx = shared_rx.lock().unwrap();
            let listener = shared_listener.lock().unwrap();
            let mut map = shared_map.lock().unwrap();
            if let Ok(e) = rx.recv() {
                println!("worker {} receive a event: {:?}", i, e);
                match e.typ {
                    EventType::Accept => {
                        // handle connection accept
                        match listener.accept() {
                            Ok((mut stream, addr)) => {
                                println!("new connection from address:{:?}", addr);
                                stream
                                    .write_all(
                                        format!("HTTP/1.1 200 OK\r\nHello!\r\n\r\n").as_bytes(),
                                    )
                                    .unwrap();
                                // add stream to epoll
                                let mut event = ffi::EpollEvent {
                                    events: (ffi::EPOLLIN | ffi::EPOLLRDHUP) as u32,
                                    data: ffi::EpollData {
                                        fd: stream.as_raw_fd(),
                                    },
                                };
                                stream.set_nonblocking(true).unwrap();
                                let ret = unsafe {
                                    ffi::epoll_ctl(
                                        epfd,
                                        ffi::EPOLL_CTL_ADD,
                                        stream.as_raw_fd(),
                                        &mut event,
                                    )
                                };
                                panic_if_error(ret);
                                // stream 在此处被析构了，所以必须将 stream 装进容器中，下次再使用
                                map.insert(stream.as_raw_fd(), stream);
                            }
                            Err(e) => println!("couldn't get client: {:?}", e),
                        }
                    }
                    EventType::IO => {
                        // 目前只监听了 read 事件
                        if let Some(stream) = map.get_mut(&e.fd) {
                            stream
                                .write_all(
                                    format!("HTTP/1.1 200 OK\r\nIO Event!\r\n\r\n").as_bytes(),
                                )
                                .unwrap();
                        }
                    }
                }
            }
        });
    }
    // let stop = false;
    loop {
        // 进入主线程逻辑，通过 epoll_wait 监听连接
        let mut events = Vec::with_capacity(10);
        // 监听最大事件个数 10，-1 表示会一直等到事件发生
        let ret = unsafe { ffi::epoll_wait(epfd, events.as_mut_ptr(), 10, -1) };
        if ret < 0 {
            break;
        }
        println!("{:} events received", ret);
        unsafe { events.set_len(ret as usize) }; // 设置有效事件个数
        for event in events {
            let fd = unsafe {
                println!("file descriptor: {:?} occur event", event.data.fd);
                event.data.fd
            };
            if (event.events & ffi::EPOLLIN as u32) == 1 {
                // 如果是 listener_fd，那么应该是新的连接
                if fd == listener_fd {
                    // send event to queue
                    sender
                        .send(Event {
                            fd,
                            typ: EventType::Accept,
                        })
                        .unwrap();
                } else {
                    // 其它情况则是IO数据
                    sender
                        .send(Event {
                            fd,
                            typ: EventType::IO,
                        })
                        .unwrap();
                }
            } else {
            }
        }
    }
    let ret = unsafe { ffi::close(epfd) }; // 关闭 epoll
    panic_if_error(ret);
    println!("FINISHED!");
}

#[derive(Debug)]
enum EventType {
    Accept,
    IO,
}

#[derive(Debug)]
struct Event {
    pub fd: i32,
    pub typ: EventType,
}

fn panic_if_error(ret: i32) {
    if ret < 0 {
        panic!(io::Error::last_os_error());
    }
}
